package com.flink.hbase.windowingoptimization.config;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.BoundedOutOfOrdernessWatermarks;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.Watermark;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;

/**
 * Watermark配置类
 * 提供不同业务场景的Watermark策略配置
 * 
 * @author Flink Developer
 */
public class WatermarkConfig {
    
    private static final Logger logger = LoggerFactory.getLogger(WatermarkConfig.class);
    
    /**
     * 低延迟场景Watermark配置
     * 适用于：用户行为实时分析、实时监控告警
     */
    public static <T> WatermarkStrategy<T> lowLatencyStrategy(
            SerializableTimestampAssigner<T> timestampAssigner) {
        return WatermarkStrategy
                .<T>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(timestampAssigner)
                .withIdleness(Duration.ofSeconds(5));
    }
    
    /**
     * 高吞吐场景Watermark配置
     * 适用于：批量数据处理、大规模日志分析
     */
    public static <T> WatermarkStrategy<T> highThroughputStrategy(
            SerializableTimestampAssigner<T> timestampAssigner) {
        return WatermarkStrategy
                .<T>forBoundedOutOfOrderness(Duration.ofMinutes(1))
                .withTimestampAssigner(timestampAssigner)
                .withIdleness(Duration.ofMinutes(2));
    }
    
    /**
     * 严格有序场景Watermark配置
     * 适用于：金融交易、支付处理
     */
    public static <T> WatermarkStrategy<T> strictOrderStrategy(
            SerializableTimestampAssigner<T> timestampAssigner) {
        return WatermarkStrategy
                .<T>forMonotonousTimestamps()
                .withTimestampAssigner(timestampAssigner)
                .withIdleness(Duration.ofSeconds(30));
    }
    
    /**
     * 混合场景Watermark配置
     * 适用于：IoT设备数据、移动应用数据
     */
    public static <T> WatermarkStrategy<T> mixedScenarioStrategy(
            SerializableTimestampAssigner<T> timestampAssigner) {
        return WatermarkStrategy
                .<T>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                .withTimestampAssigner(timestampAssigner)
                .withIdleness(Duration.ofMinutes(1));
    }
    
    /**
     * 自定义Watermark生成器
     * 支持动态调整延迟和空闲超时
     */
    public static class AdaptiveWatermarkGenerator<T> implements WatermarkGenerator<T> {
        private final Duration maxOutOfOrderness;
        private final Duration idleTimeout;
        
        private long maxTimestamp = Long.MIN_VALUE;
        private long lastEventTime = Long.MIN_VALUE;
        
        public AdaptiveWatermarkGenerator(Duration maxOutOfOrderness, Duration idleTimeout) {
            this.maxOutOfOrderness = maxOutOfOrderness;
            this.idleTimeout = idleTimeout;
        }
        
        @Override
        public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
            maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            lastEventTime = System.currentTimeMillis();
        }
        
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            long currentTime = System.currentTimeMillis();
            
            // 检查是否长时间无数据（空闲检测）
            if (currentTime - lastEventTime > idleTimeout.toMillis()) {
                logger.warn("数据源空闲超时，发送当前时间戳作为Watermark: {}", currentTime);
                output.emitWatermark(new Watermark(currentTime - maxOutOfOrderness.toMillis()));
            } else {
                // 正常情况下发送Watermark
                output.emitWatermark(new Watermark(maxTimestamp - maxOutOfOrderness.toMillis()));
            }
        }
    }
    
    /**
     * 自定义Watermark策略
     * 支持基于数据特征的动态调整
     */
    public static <T> WatermarkStrategy<T> adaptiveStrategy(
            SerializableTimestampAssigner<T> timestampAssigner,
            Duration maxOutOfOrderness,
            Duration idleTimeout) {
        
        return WatermarkStrategy
                .<T>forGenerator((context) -> new AdaptiveWatermarkGenerator<>(maxOutOfOrderness, idleTimeout))
                .withTimestampAssigner(timestampAssigner)
                .withIdleness(idleTimeout);
    }
    
    /**
     * 多级Watermark策略
     * 为不同的数据流设置不同的延迟容忍度
     */
    public static class MultiLevelWatermarkGenerator<T> implements WatermarkGenerator<T> {
        private final Duration level1Delay;  // 快速数据流
        private final Duration level2Delay;  // 中速数据流
        private final Duration level3Delay;  // 慢速数据流
        
        private long maxTimestamp = Long.MIN_VALUE;
        private String streamType;
        
        public MultiLevelWatermarkGenerator(Duration level1, Duration level2, Duration level3) {
            this.level1Delay = level1;
            this.level2Delay = level2;
            this.level3Delay = level3;
        }
        
        @Override
        public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
            maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            
            // 根据事件类型判断数据流级别
            if (event instanceof StreamTypeAware) {
                streamType = ((StreamTypeAware) event).getStreamType();
            }
        }
        
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            Duration delay = getDelayByStreamType(streamType);
            output.emitWatermark(new Watermark(maxTimestamp - delay.toMillis()));
        }
        
        private Duration getDelayByStreamType(String streamType) {
            if (streamType == null) return level2Delay;
            
            switch (streamType) {
                case "fast":
                    return level1Delay;
                case "medium":
                    return level2Delay;
                case "slow":
                    return level3Delay;
                default:
                    return level2Delay;
            }
        }
    }
    
    /**
     * 流类型感知接口
     */
    public interface StreamTypeAware {
        String getStreamType();
    }
    
    /**
     * 性能监控Watermark包装器
     */
    public static class MonitoringWatermarkGenerator<T> implements WatermarkGenerator<T> {
        private final WatermarkGenerator<T> delegate;
        private final String metricName;
        
        private long watermarkCount = 0;
        private long lastWatermarkTime = 0;
        
        public MonitoringWatermarkGenerator(WatermarkGenerator<T> delegate, String metricName) {
            this.delegate = delegate;
            this.metricName = metricName;
        }
        
        @Override
        public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
            delegate.onEvent(event, eventTimestamp, new MonitoringWatermarkOutput(output));
        }
        
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            delegate.onPeriodicEmit(new MonitoringWatermarkOutput(output));
        }
        
        private class MonitoringWatermarkOutput implements WatermarkOutput {
            private final WatermarkOutput delegate;
            
            public MonitoringWatermarkOutput(WatermarkOutput delegate) {
                this.delegate = delegate;
            }
            
            @Override
            public void emitWatermark(Watermark watermark) {
                watermarkCount++;
                lastWatermarkTime = watermark.getTimestamp();
                
                // 记录Watermark指标
                if (watermarkCount % 1000 == 0) {
                    logger.info("Watermark统计 [{}]: 总数={}, 最新时间戳={}", 
                        metricName, watermarkCount, lastWatermarkTime);
                }
                
                delegate.emitWatermark(watermark);
            }
            
            @Override
            public void markIdle() {
                logger.info("数据源变为空闲状态: {}", metricName);
                delegate.markIdle();
            }
            
            @Override
            public void markActive() {
                logger.info("数据源变为活跃状态: {}", metricName);
                delegate.markActive();
            }
        }
    }
    
    /**
     * 配置参数类
     */
    public static class WatermarkConfigParams {
        private Duration maxOutOfOrderness = Duration.ofSeconds(30);
        private Duration idleTimeout = Duration.ofMinutes(1);
        private boolean enableMonitoring = true;
        private String metricPrefix = "watermark";
        
        // Getters and Setters
        public Duration getMaxOutOfOrderness() { return maxOutOfOrderness; }
        public void setMaxOutOfOrderness(Duration maxOutOfOrderness) { this.maxOutOfOrderness = maxOutOfOrderness; }
        
        public Duration getIdleTimeout() { return idleTimeout; }
        public void setIdleTimeout(Duration idleTimeout) { this.idleTimeout = idleTimeout; }
        
        public boolean isEnableMonitoring() { return enableMonitoring; }
        public void setEnableMonitoring(boolean enableMonitoring) { this.enableMonitoring = enableMonitoring; }
        
        public String getMetricPrefix() { return metricPrefix; }
        public void setMetricPrefix(String metricPrefix) { this.metricPrefix = metricPrefix; }
    }
    
    /**
     * 从配置文件加载Watermark参数
     */
    public static WatermarkConfigParams loadFromProperties(Properties props) {
        WatermarkConfigParams config = new WatermarkConfigParams();
        
        String maxOutOfOrderness = props.getProperty("watermark.max.out.of.orderness", "30s");
        config.setMaxOutOfOrderness(Duration.parse("PT" + maxOutOfOrderness.toUpperCase()));
        
        String idleTimeout = props.getProperty("watermark.idle.timeout", "1m");
        config.setIdleTimeout(Duration.parse("PT" + idleTimeout.toUpperCase()));
        
        boolean enableMonitoring = Boolean.parseBoolean(
            props.getProperty("watermark.monitoring.enabled", "true"));
        config.setEnableMonitoring(enableMonitoring);
        
        String metricPrefix = props.getProperty("watermark.metric.prefix", "watermark");
        config.setMetricPrefix(metricPrefix);
        
        return config;
    }
    
    /**
     * 工厂方法：创建监控包装的Watermark策略
     */
    public static <T> WatermarkStrategy<T> createMonitoringStrategy(
            WatermarkStrategy<T> baseStrategy, 
            String metricName) {
        
        return WatermarkStrategy
                .<T>forGenerator((context) -> {
                    WatermarkGenerator<T> baseGenerator = baseStrategy.createWatermarkGenerator(context);
                    return new MonitoringWatermarkGenerator<>(baseGenerator, metricName);
                })
                .withTimestampAssigner(baseStrategy.createTimestampAssigner(null))
                .withIdleness(Duration.ofMinutes(1));
    }
} 